package com.eastcom.dataquality.dv.analysis;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import com.eastcom.dataquality.dv.CollectData;
import com.eastcom.dataquality.dv.analysis.extract.Extractor;

public class IntegerWorker extends Thread{

	private int capacity;
	
	private BlockingQueue<String> queue;
	
	private List<Extractor> extractors;
	
	private static Logger logger=Logger.getLogger(IntegerWorker.class);
	
	private long processCount;
	
	public IntegerWorker(int capacity) {
		super();
		this.capacity = capacity;
	}
	
	public void init(){
		queue=new LinkedBlockingQueue(capacity);
	}
	
	public boolean enterData(String data){
		logger.info("IntegerWorker enterData");
		return queue.offer(data);
	}

	@Override
	public void run(){
		while(true){
			try {
				String cltData=queue.take();
				CollectData collectData=OriginDataTransFormator.transFormFrom(cltData);
				for(Extractor extractor:extractors){
					extractor.process(collectData);
				}
				logger.info(this.getName()+"have process "+(++processCount)+" collectData");
			} catch (Exception e) {
				e.printStackTrace();
				logger.error("IntegerWorker.run() have occure exception for:"+e.getMessage());
			}
		}
	}
	
	public List<Extractor> getExtractors() {
		return extractors;
	}

	public void setExtractors(List<Extractor> extractors) {
		this.extractors = extractors;
	}
}
